#define _GNU_SOURCE
#include "app_pool.h"
#include <mqueue.h>
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <sys/types.h>
#include "log/log.h"
#include <unistd.h>

static char *mq_name = "/mq_pool"; // /dev/mqueue下的文件
static mqd_t mqd; // 队列描述符
static pthread_t *pthreads; // 用来存储所有创建的线程标识的指针
static int pool_size; // 线程池大小

// 不断的从队列中读取任务，如果得到了，执行其中任务函数， 否则等待
void * thread_func(void *arg)
{
    log_debug("thread_func start, tid=%d", gettid());

    Task task;
    while(1)
    {
        ssize_t len = mq_receive(mqd, (char *)&task, sizeof(Task), NULL); // 从队列中读取任务
        if (len == sizeof(Task)) // 接收的确实是task
        {
            task.func(task.arg);
        }
    }
}

/**
 * 初始化线程池（和队列）
 */
int app_pool_init(int size)
{
    // 创建任务队列
    struct mq_attr attr;
    attr.mq_maxmsg = 10; // 队列中最大消息数
    attr.mq_msgsize = sizeof(Task); // 每条消息的最大大小
    mqd = mq_open(mq_name, O_CREAT | O_RDWR, 0644, &attr);
    if (mqd == -1)
    {
        log_error("mq_open error");
        return -1;
    }

    // 创建线程池
   pthreads = malloc(sizeof(pthread_t) * size);
   for (int i = 0; i < size; i++)
   {
        if (pthread_create(&pthreads[i], NULL, thread_func, NULL) == -1)
        {
            log_error("pthread_create error");
            free(pthreads);
            mq_unlink(mq_name);
            return -1;
        }
   }
   pool_size = size;

   return 0;
}

/**
 * 关闭线程池（和队列）
 */
void app_pool_close()
{
    // 取消结束线程池中的所有线程
    for (int i = 0; i < pool_size; i++)
    {
        pthread_cancel(pthreads[i]); // 取消线程
        pthread_join(pthreads[i], NULL); // 等待线程结束
    }

    // 释放线程池
    free(pthreads);

    // 关闭队列
    mq_unlink(mq_name);
}

/**
 * 向添加(注册)任务（将任务发送到队列）
 */
int app_pool_registerTask(int (*func)(void *), void *arg)
{
    // 创建任务
    Task task = {
        .func = func,
        .arg = arg
    };
    // Task *task2 = malloc(sizeof(Task));
    // task2->func = func;
    // task2->arg = arg;

    // 发送任务
    mq_send(mqd, (char *)&task, sizeof(Task), 0);
}